package com.zcy;

import com.espertech.esper.common.client.EPCompiled;
import com.espertech.esper.common.client.EventBean;
import com.espertech.esper.common.client.configuration.Configuration;
import com.espertech.esper.common.client.util.SafeIterator;
import com.espertech.esper.compiler.client.CompilerArguments;
import com.espertech.esper.compiler.client.EPCompileException;
import com.espertech.esper.compiler.client.EPCompiler;
import com.espertech.esper.compiler.client.EPCompilerProvider;
import com.espertech.esper.runtime.client.*;
import com.zcy.event.MyEvent;
import com.zcy.event.subscribe.MyEventSubscribe;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

/**
 * 三种方式获取数据
 */
@Slf4j
public class ThreeReceivingMethods {

    private String epl = "@name('subscribe') select * from MyEvent;";

    /**
     * Subscribe
     * 也是官网推荐的一种
     * 其相较于Listener少了一层转换成EventType的方式push出来
     * 但每个语句只能添加一个Subscribe，但可以添加多个Listener
     * <p>
     * 关于Subscribe中的方法，Esper 内部会通过反射的方式去选择最优的Subscribe的方法，找到之后反射的调用该方法
     * 具体源码详见：{@link new com.espertech.esper.runtime.internal.subscriber.ResultDeliveryStrategyFactory#create}
     * <p>
     * 反射创建ResultDeliveryStrategyImpl只会反射一次
     *
     * @throws EPCompileException
     * @throws EPDeployException
     * @throws InterruptedException
     */
    @Test
    public void subscribe() throws EPCompileException, EPDeployException, InterruptedException {
        EPCompiler compiler = EPCompilerProvider.getCompiler();


        Configuration configuration = new Configuration();
        configuration.getCompiler().getByteCode().setAllowSubscriber(true);
        configuration.getCommon().addEventType(MyEvent.class);

        CompilerArguments args = new CompilerArguments(configuration);


        EPCompiled compiled = compiler.compile(epl, args);

        EPRuntime defaultRuntime = EPRuntimeProvider.getDefaultRuntime(configuration);
        EPDeploymentService deploymentService = defaultRuntime.getDeploymentService();
        EPDeployment deploy = deploymentService.deploy(compiled);


        EPEventService eventService = defaultRuntime.getEventService();

        EPStatement subscribeStatement = deploymentService.getStatement(deploy.getDeploymentId(), "subscribe");

        subscribeStatement.setSubscriber(new MyEventSubscribe());

        // 发布之前的数据是接受不到的，因此在添加监听器或者订阅者的时候应该尽早进行
        eventService.sendEventBean(MyEvent.builder().amount("0001").build(), "MyEvent");
        Thread.sleep(2000);
        eventService.sendEventBean(MyEvent.builder().amount("0001").build(), "MyEvent");

        Thread.sleep(1000000);
    }


    @Test
    public void listener() throws EPCompileException, EPDeployException, InterruptedException {
        EPCompiler compiler = EPCompilerProvider.getCompiler();


        Configuration configuration = new Configuration();
        configuration.getCompiler().getByteCode().setAllowSubscriber(true);
        configuration.getCommon().addEventType(MyEvent.class);

        CompilerArguments args = new CompilerArguments(configuration);


        EPCompiled compiled = compiler.compile(epl, args);

        EPRuntime defaultRuntime = EPRuntimeProvider.getDefaultRuntime(configuration);
        EPDeploymentService deploymentService = defaultRuntime.getDeploymentService();
        EPDeployment deploy = deploymentService.deploy(compiled);


        EPEventService eventService = defaultRuntime.getEventService();

        EPStatement subscribeStatement = deploymentService.getStatement(deploy.getDeploymentId(), "subscribe");

        subscribeStatement.addListener((newEvents, oldEvents, statement, runtime) -> log.info(""));

        // 发布之前的数据是接受不到的，因此在添加监听器或者订阅者的时候应该尽早进行
        eventService.sendEventBean(MyEvent.builder().amount("0001").build(), "MyEvent");
        Thread.sleep(2000);
        eventService.sendEventBean(MyEvent.builder().amount("0001").build(), "MyEvent");

        Thread.sleep(1000000);
    }

    /**
     * 这是Pull方式
     * TODO 试验不成功
     *
     * @throws EPCompileException
     * @throws EPDeployException
     * @throws InterruptedException
     */
    @Test
    public void safeIterator() throws EPCompileException, EPDeployException, InterruptedException {
        EPCompiler compiler = EPCompilerProvider.getCompiler();


        Configuration configuration = new Configuration();
        configuration.getCompiler().getByteCode().setAllowSubscriber(true);
        configuration.getCommon().addEventType(MyEvent.class);

        CompilerArguments args = new CompilerArguments(configuration);


        EPCompiled compiled = compiler.compile(epl, args);

        EPRuntime defaultRuntime = EPRuntimeProvider.getDefaultRuntime(configuration);
        EPDeploymentService deploymentService = defaultRuntime.getDeploymentService();
        EPDeployment deploy = deploymentService.deploy(compiled);


        EPEventService eventService = defaultRuntime.getEventService();

        EPStatement subscribeStatement = deploymentService.getStatement(deploy.getDeploymentId(), "subscribe");

        eventService.sendEventBean(MyEvent.builder().amount("0001").build(), "MyEvent");
        eventService.sendEventBean(MyEvent.builder().amount("0001").build(), "MyEvent");
        eventService.sendEventBean(MyEvent.builder().amount("0001").build(), "MyEvent");
        eventService.sendEventBean(MyEvent.builder().amount("0001").build(), "MyEvent");
        eventService.sendEventBean(MyEvent.builder().amount("0001").build(), "MyEvent");


        // 发布之前的数据是接受不到的，因此在添加监听器或者订阅者的时候应该尽早进行
        SafeIterator<EventBean> eventBeanSafeIterator = null;
        try {
            eventBeanSafeIterator = subscribeStatement.safeIterator();
            while (eventBeanSafeIterator.hasNext()) {
                EventBean next = eventBeanSafeIterator.next();
                log.info(next.toString());
            }
        } finally {
            if (eventBeanSafeIterator != null) {
                eventBeanSafeIterator.close();
            }
        }

        eventService.sendEventBean(MyEvent.builder().amount("0001").build(), "MyEvent");


        Thread.sleep(1000000);
    }


}